package Top4.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import util.ConnectionUtil;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * @Auther: Administrator
 * @Date: 2019/3/5 22:04
 * @Description:
 */
public class SendAsync {
    private static final String QUEUE_NAME = "queue_confirm";

    public static void main(String[] args) throws IOException, TimeoutException {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();
        //创建通道
        Channel channel = connection.createChannel();
        //声明一个持久化的 队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //开启confirm模式
        channel.confirmSelect();

        //存储未确认的消息标识tag
        final SortedSet<Long> confirmSet = Collections.synchronizedNavigableSet(new TreeSet<Long>());

        //开启监听
        channel.addConfirmListener(new ConfirmListener() {
            /**
             * 处理返回确认成功
             * @param deliveryTag 如果是多条，这个就是最后一条消息的tag
             * @param multiple 是否多条
             * @throws IOException
             */
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {

                System.out.println("消息发送成功,deliveryTag:" + deliveryTag + "multiple:" + multiple + "");
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }

            /**
             * 处理返回确认失败
             * @param deliveryTag
             * @param multiple
             * @throws IOException
             */
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("失败,deliveryTag:" + deliveryTag + "multiple:" + multiple + "");
                if (multiple) {

                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }
        });

        try {
            for (int i = 0; 1 < 2; i++) {
                String msg = "我是confirm模式 消息 异步【" + i + "】";
                long tag = channel.getNextPublishSeqNo();
                //发送消息
                channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
                System.out.println("tag:" + tag);
                confirmSet.add(tag);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            channel.close();
            connection.close();
        }

        //上面是无限循环所以这个应该放到finally里
        //
    }

}
